package lj.netty.listen;

import cn.hutool.core.lang.Opt;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import cn.hutool.log.StaticLog;
import io.netty.handler.codec.ByteToMessageDecoder;
import lj.a.protocol.lj.bean.LJProtocolInfo;
import lj.netty.bean.LJNettyBo;
import lj.netty.bean.LJNettyMsgContext;
import lj.netty.tool.LJTNetty;
import lj.netty.tool.LJTNettyStructPackage;
import lombok.NoArgsConstructor;
import ntd.remote.link.ComRegistryClient;
import ntd.remote.online.ComOnline;
import ntd.remote.online.bo.OnlineInfo;
import ntd.util.ToolCache;
import ntd.util.UtilCatch;
import ntd.util.fun.ComFun;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

@NoArgsConstructor
public abstract class LJNettyListen implements LJCNettyListen {

    public static final String SEND_ALL = "*";

    private static final LJTNettyStructPackage ljtNettyStructPackage = new LJTNettyStructPackage("LJLJLJLJLJ");

    private Map<String, ComFun.FunIn<LJNettyMsgContext>> funInHashMap = new HashMap<>();

    public abstract byte[] onLjMsg(LJNettyMsgContext ljNettyMsgContext);

    public LJNettyMsgContext pushMsgAll(LJProtocolInfo ljProtocolInfo) {
        // 发送报文
        LJTNetty.pushAll(ljProtocolInfo.toBytes());
        return awaitRes(ljProtocolInfo.getMsgCode());
    }

    public LJNettyMsgContext pushMsg(String nettyCode, LJProtocolInfo ljProtocolInfo) {
        LJTNetty.push(nettyCode, ljProtocolInfo.toBytes());
        return awaitRes(ljProtocolInfo.getMsgCode());
    }

    private LJNettyMsgContext awaitRes(String msgCode) {
        // 等待响应
        CountDownLatch countDownLatch = ThreadUtil.newCountDownLatch(1);
        AtomicReference<LJNettyMsgContext> contextAtomicReference = new AtomicReference<>();
        funInHashMap.put(msgCode, ljNettyMsgContext -> {
//            StaticLog.info("完成等待函数[{}]", msgCode);
            contextAtomicReference.set(ljNettyMsgContext);
            countDownLatch.countDown();
        });
//        StaticLog.info("创建等待函数[{}]", msgCode);
        // 设置超时释放 - 60秒
        UtilCatch.tryCatchNoRes(() -> countDownLatch.await(60 * 1000, TimeUnit.MILLISECONDS));
        // 返回数据
        return contextAtomicReference.get();
    }

    /**
     * 回复消息
     * @param nettyCode
     * @param msgCode
     */
    public void reply(String nettyCode, String msgCode, byte[] bytes) {
        LJProtocolInfo ljProtocolInfo = new LJProtocolInfo();
        ljProtocolInfo.setMsgCode(msgCode);
        ljProtocolInfo.setReply(true);
        ljProtocolInfo.setBytes(bytes);
        LJTNetty.push(nettyCode, ljProtocolInfo.toBytes());
    }

    /**
     * 接入消息通道
     * @param LJNettyBo
     */
    public final void onMsg(LJNettyBo LJNettyBo) {
        LJProtocolInfo ljProtocolInfo = LJProtocolInfo.parse(LJNettyBo.getBytes());

        LJNettyMsgContext ljNettyMsgContext = new LJNettyMsgContext();
        ljNettyMsgContext.setLJNettyBo(LJNettyBo);
        ljNettyMsgContext.setLjProtocolInfo(ljProtocolInfo);

        String key = ToolCache.get("client-code");
        if (StrUtil.isBlank(key)) {
            ToolCache.set("client-code", "*");
            Opt.ofNullable(ComRegistryClient.ljNettyInitContext).ifPresent(ljNettyInitContext -> {
                ToolCache.set("client-code", ljNettyInitContext.getTerminalCode());
            });
        }

        if (ljProtocolInfo.isReply()) {
            // 回复消息处理逻辑
            Opt.ofNullable(funInHashMap.remove(ljProtocolInfo.getMsgCode())).ifPresent(funIn -> UtilCatch.tryCatchNoRes(() -> funIn.accept(ljNettyMsgContext)));
        } else if (StrUtil.isNotBlank(ljProtocolInfo.getGoalTerminal()) && !key.equals(ljProtocolInfo.getGoalTerminal())) {
            // 目标转发报文
            OnlineInfo onlineInfo = ComOnline.getOnlineInfo("appInfo", ljProtocolInfo.getGoalTerminal());
            LJNettyMsgContext ljNettyMsgContextT = pushMsg(onlineInfo.getNettyChannelCode(), ljProtocolInfo);
            reply(LJNettyBo.getNettyCode(), ljProtocolInfo.getMsgCode(), ljNettyMsgContextT.getLjProtocolInfo().getBytes());
        } else {
            UtilCatch.tryCatchNoRes(() -> {
                // 执行逻辑
                byte[] accept = onLjMsg(ljNettyMsgContext);
                // 回复消息
                reply(LJNettyBo.getNettyCode(), ljProtocolInfo.getMsgCode(), accept);
            });
        }
    }

    @Override
    public void onClose(LJNettyBo LJNettyBo) {

    }

    @Override
    public ByteToMessageDecoder struct() {
        return ljtNettyStructPackage.struct();
    }

    @Override
    public void onInit(LJNettyBo LJNettyBo) {

    }

    public static void main(String[] args) {
        StaticLog.info(JSONUtil.toJsonStr("LJ".getBytes()));
    }
}
